package com.ibm.cps.processors.vector;

import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;
import org.bson.types.ObjectId;
import org.joda.time.DateTime;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.processors.BasicProcessor;
import com.ibm.cps.processors.Info;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.DataTypes;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

@Info(Name=MessageFields.VECTOR_PROJECTION,Type=Info.ProcessorType.Other)
public class VectorProjectionProcessor extends BasicProcessor {
	private static final Logger logger = Logger.getLogger(VectorProjectionProcessor.class);
	
	@ValueParam @IgnoredInFunctionCall
	private String fieldMappingJSON;
	private Map<String, String> fieldNameMapping = new HashMap<>();
	private ITimeSeriesSchema parentSchema;
	private ITimeSeriesSchema outputSchema;
	
	public VectorProjectionProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
		this.parentSchema = parentSchema;
	}
	
	@Override
	public void open() throws CPSException {
		this.fieldNameMapping = getFieldMapping();
		this.outputSchema = parseSchemaFromMetadata();
	}

	private Map<String, String> getFieldMapping() throws CPSException{
		ObjectNode fieldMapping = (ObjectNode)JSONUtility.fromString(fieldMappingJSON);
		Map<String, String> result = new HashMap<>();
		Iterator<String> inputFields = fieldMapping.fieldNames();
		while(inputFields.hasNext()){
			String inputFieldName = inputFields.next();
			String outputFieldName = fieldMapping.get(inputFieldName).asText();
			result.put(inputFieldName, outputFieldName);
		}
		return result;
	}
	
	private ITimeSeriesSchema parseSchemaFromMetadata() throws CPSException{
		ArrayNode outputSchema = JSONUtility.newArrayNode();
		byte[] parentFields = parentSchema.getDataTypes();

		for(String inputFieldName : fieldNameMapping.keySet()){
			String outputFieldName = fieldNameMapping.get(inputFieldName);
			int fieldIndex = parentSchema.getTagId(inputFieldName);
			if(fieldIndex == -1){
				throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Cannot find field " + inputFieldName + " in parentSchema in projection");
			}
			String fieldType = DataTypes.getDataType(parentFields[fieldIndex]);
			ObjectNode fieldObject = JSONUtility.newObjectNode();
			fieldObject.put(MessageFields.NAME, outputFieldName);
			fieldObject.put(MessageFields.DATATYPE, fieldType);
			// refactor here to avoid add tskey and timestamp field again
			if(fieldIndex == parentSchema.getKeyIndex()){
				fieldObject.put(MessageFields.ISKEY, true);
			}
			if(fieldIndex == parentSchema.getTimestampIndex()){
				fieldObject.put(MessageFields.ISTIME, true);
			}
			outputSchema.add(fieldObject);
		}
		String schemaId = new BigInteger(new ObjectId().toByteArray()).toString(16);
		long version = (new DateTime()).getMillis();
		return new TimeSeriesSchema(tenantId, schemaId, outputSchema, version);
	}

	@Override
	public ITimeSeriesSchema getTimeSeriesSchema() throws CPSException {
		return outputSchema;
	}

	@Override
	public Collection<ParsedDataMessage> execute(ParsedDataMessage message)
			throws CPSException {
		Comparable tskey = message.getTsKey();
		long timestamp = message.getTimestamp();
		ParsedDataMessageBuilder builder = ParsedDataMessageBuilder.createBasicBuilder(outputSchema, tskey, timestamp);
		
		for (String usedInputField : fieldNameMapping.keySet()) {
			String outputFieldName = fieldNameMapping.get(usedInputField);
			int index = parentSchema.getTagId(usedInputField);
			byte typeByte = parentSchema.getDataType(index);
			Object value;
			if(typeByte == DataTypes.DT_DOUBLE_ARRAY || typeByte == DataTypes.DT_INT_ARRAY || typeByte == DataTypes.DT_STRING_ARRAY)
				value = message.getArrayValue(index);
			else
				value = message.getValue(index);
			builder.append(outputFieldName, value);
		}
		List<ParsedDataMessage> result = new ArrayList<>();
        ParsedDataMessage parsedData = builder.toParsedDataMessage();
        result.add(parsedData);
		logger.info("==============Projection output============");
		logger.info(Arrays.toString(outputSchema.getTagNames()));
		logger.info(Arrays.toString(parsedData.getObjectValues()));
		return result;
	}

	@Override
	public void dispose() {
	}


}
